Please fix import ordering, see from : https://gobblin.apache.org/docs/developer-guide/CodingStyle/

Since this function is validating whether or not the flow should be created based on whether it's a duplicate adhoc flow within a short span, can this be named differently to convey it's purpose like validateConcurrentAdhocFlowCreation ?

updated the name appropriately.

No assertions? Also in this unit test we are mocking leaseArbiter, so I don't think it adds any value on testing the functionality of leaseArbiter

Thanks for pointing out missed adding the Assertion for this test, purpose of this test is to verify that MySQLStatemanagementStore returns the value returned by leaseArbiter, also updated this test.

Just an FYI, this also gets called during updating a flow. But since we have a condition of checking the flow is scheduled or not and we don't expect users to update an adhoc flow, we should be fine.

But since this can still be called for adhoc flows, it would be good to test what the behaviour is. No need to handle it specially, but to know what the behaviour is would be good.

Thanks for the callout, will add it for the test suite

consider renaming this to isLeaseAcquirable for conciseness and to be consistent with other method names

Please add javadoc.. something like:
Determines if a lease can be acquired for the given flow. A lease is acquirable if ...

add an info log here with flowGroup, flowName.. it would be useful in debugging

Added lease params which contains details of flow name and flow group

add javadoc

updated

use camelcase for naming methods

done

can we use this.flowSpec here since it already has the schedule?

We need different configs in flowspec depending on the case we are testing and hence would prefer to define new ones.

let's also test the scenario when canAcquireLeaseOnEntity returns true for adhoc flow since the other test is for scheduled flow

beyond clearly naming for callers, impl-wise, this definitely relates to a flow, so that should be a ctor param. consider whether to allow a catcher to reach in to access the details as instance member(s) or merely to use internally in the ctor, to contextualize the message passed along to super.

very reasonable method-level javadoc... but it turns out epsilon is not mentioned anywhere in class-level javadoc, so this method description lacks context.

so, please add the class-level info. mentioning the name 'epsilon' is fine, but definitely also give it a more specific name, like "Lease Consolidation Period".

updated

the whole idea is that a "similar" (but NOT same) lease isn't itself already within epsilon. hence, be sure to test LeaseParams that were NOT just given to tryAcquireLease

created new launch param launchLeaseParams3_similar

usually exception messages are designed for logging, more than for end-user consumption, so probably not appropriate to blindly return that. (it's sometimes done for a 5xx error, as above... but even that can be inadvisable.)

anyway, the 409 above might offer a better template:

return new CreateKVResponse<>(new RestLiServiceException(HttpStatus.S_409_CONFLICT,
    "FlowSpec with URI " + flowSpec.getUri() + " was launched less than N secs ago, no action will be taken"));

(to provide N we may wish to tunnel the value of epsilon... or at least how many secs remain before a subsequent launch would be possible)

also: when do we want to return (as that 409 above does), vs. throw?

updated as discussed offline

javadoc seems out-of-date, esp. mentioning LeaseParams and DagAction

also, out-of-date:

Returns true if lease can be acquired on entity provided in leaseParams.

updated javadoc

sense is reversed here...

maybe:

Check whether the same flowGroup+flowName is within the Lease Consolidation Period (aka. epsilon) from other, unrelated leasing activity

this is also out-of-date:

@return true if lease can be acquired on the flow passed in the lease params, false otherwise

updated javadoc

1. we have an .info line above announcing the check, so let's follow w/ a .warn line here when the check fails. suggest: "another recent adhoc flow exec found for...."

2. exception msg could benefit from minor improvements, yet--however it's phrased--it belongs encapsulated in the TooSoonToRerun... ctor, which should take solely a FlowSpec param

Updated, Thanks for the suggestion

I never noticed before that flowExecutionId, which is customarily millis-since-epoch only has 7 digits when it should have 10. let's fix that and also define this as:

private static final long flowExecutionIdAlt = flowExecutionId + ...; // whatever you consider a reasonable (later) offset

updated the same variables to store millis

the cast isn't necessary (the reason I suggested wrapping TooSoonToR... was to enable uniform, type-agnostic code here.)

for a method with the signature throws Throwable, aren't these two equivalent?

throw error.getCause();
throw (T) error.getCause();

makes sense updated

after signing off, I realized my literal advice would compromise clarity, foul up tests, etc. (in this very way)

sorry for that half-baked advice... try this instead:

public static class TooSoonToRerunSameFlowException extends RuntimeException {
  @Getter private final FlowSpec flowSpec;

  /**
   * Account for unwrapping within @{link FlowCatalog#updateOrAddSpecHelper}`s `CallbackResult` error handling for `SpecCatalogListener`s
   * @return `TooSoonToRerunSameFlowException` wrapped in another `TooSoonToRerunSameFlowException
   */
  public static TooSoonToRerunSameFlowException wrappedOnce(FlowSpec flowSpec) {
    return new TooSoonToRerunSameFlowException(flowSpec, new TooSoonToRerunSameFlowException(flowSpec));
  }

  public TooSoonToRerunSameFlowException(FlowSpec flowSpec) {
    this(flowSpec, null);
  }

  /** restricted-access ctor: use {@link #wrappedOnce(String)} instead */
  private TooSoonToRerunSameFlowException(FlowSpec flowSpec, Throwable cause) {
    super("Lease already occupied by another recent execution of this flow: " + flowSpec, cause);
    this.flowSpec = flowSpec;
  }
}

then replace:

throw new RuntimeException(new TooSoonToRerunSameFlowException(flowSpec));

with

throw TooSoonToRerunSameFlowException.wrappedOnce(flowSpec);

Sure updated as per suggestion

probably no need for this comment here in the impl, but if you want one, bring it into line w/ the orig from the interface

makes sense, since javadoc already there for the interface, removed comment from here

this import belongs a few lines down w/ other apache gobblin pkgs

Updated

nit: {@link TooSoonToRerunSameFlowException}

Sure updated

forgot to change this back to TooSoonToRerun...

my bad, updated

needs update

updated javadoc